#include "PahoMqttClient.h"
#include <stdio.h>
#include <string.h>
#include <limits.h>  // 包含 limits.h 以使用常量
#include <debug_printf.h>
#include "thmar_types.h"
#include <stdlib.h>
#include "local_error.h"
#include "c11code.h"
#include "sds.h"

#define RETURN_ERR_IF_UNINITIALIZED if(this->initialized != THMR_OK)return THMR_MQTT_INIT_FAILED;
#define RETURN_IF_UNINITIALIZED if(this->initialized != THMR_OK)return;

namespace smq {

typedef struct error_info {
    int code;
    const char* message;
} error_info_t;
// paho mqtt 错误码及其对应的错误信息
static error_info_t error_infos[] = {
    {MQTTCLIENT_SUCCESS, "success"},
    {MQTTCLIENT_FAILURE, "failure"},
    {MQTTCLIENT_DISCONNECTED, "disconnected"},
    {MQTTCLIENT_MAX_MESSAGES_INFLIGHT, "max messages inflight"},
    {MQTTCLIENT_BAD_UTF8_STRING, "bad utf8 string"},
    {MQTTCLIENT_NULL_PARAMETER, "null parameter"},
    {MQTTCLIENT_TOPICNAME_TRUNCATED, "topicname truncated"},
    {MQTTCLIENT_BAD_STRUCTURE, "bad structure"},
    {MQTTCLIENT_BAD_QOS, "bad qos"},
    {MQTTCLIENT_SSL_NOT_SUPPORTED, "ssl not supported"},
    {MQTTCLIENT_BAD_MQTT_VERSION, "bad mqtt version"},
    {MQTTCLIENT_BAD_PROTOCOL, "bad protocol"},
    {MQTTCLIENT_BAD_MQTT_OPTION, "bad mqtt option"},
    {MQTTCLIENT_WRONG_MQTT_VERSION, "wrong mqtt version"},
    {MQTTCLIENT_0_LEN_WILL_TOPIC, "0 len will topic"}
};

// 返回 paho mqtt 错误码对应的错误信息
static const char* get_error_message(int code) {
    for(int i=0; i<sizeof(error_infos)/sizeof(error_info_t); i++) {
        if(error_infos[i].code == code) {
            return error_infos[i].message;
        }
    }
    return "unknown error";
}
// 根据 paho mqtt 错误码设置本地错误码及其对应的错误信息
static void set_local_error_message(int paho_code) {
    thmar_set_err(paho_code,get_error_message(paho_code));
}
// 打印 mqtt 消息负载, 仅用于调试
// payload指针指向的消息负载不一定是以null结尾的, 所以不能用printf打印
// 需要使用sds加'\0'的方式打印,sds分配的内存在打印后释放
static void print_playload(const char *payload, int payloadlen)
{
    if (payload && payloadlen > 0)
    {
        sds s = sdsnewlen(payload, payloadlen);
        printf("payload: %s\n", s);
        sdsfree(s);
    }
}

static void connectionLost(void *context, char *cause)
{
    PahoMqttClient* pahoClient = (PahoMqttClient*)context;

    COM_PRINTF_DEBUG("Connection lost\n");
    if(cause)
        COM_PRINTF_DEBUG("cause: %s\n", cause);
    int code;
    int tryCountLimit = 10;
    while((code = pahoClient->connect()) != THMR_OK && tryCountLimit-- > 0)
    {
        thmar_sleep_ms(5000);
        COM_PRINTF_DEBUG("Try to reconnect again in 5 seconds (try count limit: %d)\n", tryCountLimit);
    }
    if(code!= THMR_OK) {
        COM_PRINTF_DEBUG("Reconnect failed\n");
        thmar_print_local_err();
        exit(1);
    }
    pahoClient->subscribe("test_paho");
    COM_PRINTF_DEBUG("Auto reconnect success\n");
}
static int messageArrived(void* context, char* topicName, int topicLen, MQTTClient_message* message){
    // printf("Message arrived\n");
    // printf("topic: %s\n", topicName);
    // print_playload((char*)message->payload, message->payloadlen);
    COM_PRINTF_DEBUG("Message arrived, topic: %s(%d), message: %s\n", topicName, topicLen, (char*)message->payload);
    PahoMqttClient* pahoClient = (PahoMqttClient*)context;
    if(pahoClient->channelDispatcher) {
        COM_PRINTF_DEBUG("call dispatcher function %p\n",pahoClient->channelDispatcher);
        pahoClient->channelDispatcher(pahoClient->dispatcherContext, topicName,(size_t)topicLen, (char*)message->payload, (size_t)message->payloadlen);
    }
    MQTTClient_freeMessage(&message);
    MQTTClient_free(topicName);
    return 1;
}

static void wait_for_conn(PahoMqttClient* pahoClient)
{
    while(MQTTClient_isConnected(pahoClient->client) == 0)
    {
        thmar_sleep_ms(100);
    }
}

static int init(PahoMqttClient* pahoClient)
{
    pthread_mutex_init(&pahoClient->publish_mutex,nullptr);
    pthread_mutex_init(&pahoClient->subscribe_mutex,nullptr);

    int code = MQTTClient_create(&pahoClient->client, pahoClient->serverURI, pahoClient->clientID, MQTTCLIENT_PERSISTENCE_NONE, nullptr);
    if(code != MQTTCLIENT_SUCCESS){
        set_local_error_message(code);
        return THMR_MQTT_INIT_FAILED;
    }
    code = MQTTClient_setCallbacks(pahoClient->client, pahoClient, connectionLost, messageArrived, nullptr);
    if(code != MQTTCLIENT_SUCCESS){
        set_local_error_message(code);
        return THMR_MQTT_INIT_FAILED;
    }
    return THMR_OK;
}
static void deinit(PahoMqttClient* pahoClient)
{
    MQTTClient_disconnect(pahoClient->client, 10000);
    MQTTClient_destroy(&pahoClient->client);
    free((void*)pahoClient->serverURI);
    pahoClient->serverURI = nullptr;
    free((void*)pahoClient->clientID);
    pahoClient->clientID = nullptr;
    if(pahoClient->conn_opts.username){
        free((void*)pahoClient->conn_opts.username);
        pahoClient->conn_opts.username = nullptr;
    }
    if(pahoClient->conn_opts.password){
        free((void*)pahoClient->conn_opts.password);
        pahoClient->conn_opts.password = nullptr;
    }
    pthread_mutex_destroy(&pahoClient->publish_mutex);
    pthread_mutex_destroy(&pahoClient->subscribe_mutex);
}

PahoMqttClient::PahoMqttClient(const MQTTClient_connectOptions& conn_opts,const char* serverURI,const char *clientID,int qos)
        : ThmarMqttClient(), client(nullptr), conn_opts(conn_opts),qos(qos),initialized(INT_MAX) {
    this->serverURI = serverURI ? strdup(serverURI): strdup("tcp://localhost:1883") ;
    this->clientID = clientID ? strdup(clientID): strdup("ThmarMqttClient") ;
    if(this->conn_opts.username)
        this->conn_opts.username = strdup(this->conn_opts.username);
    if(this->conn_opts.password)
        this->conn_opts.password = strdup(this->conn_opts.password);
    this->initialized = init(this);
}

PahoMqttClient::PahoMqttClient(const char *serverURI, const char *username, const char *password, const char *clientID, int timeout, int keepAliveInterval, int qos)
    : PahoMqttClient({ {'M', 'Q', 'T', 'C'}, 8, keepAliveInterval, 1, 1, \
        NULL, username, password, timeout, 0, NULL,\
        0, NULL, MQTTVERSION_DEFAULT, {NULL, 0, 0}, {0, NULL}, -1, 0, NULL, NULL, NULL},
        serverURI,clientID,qos)
{
}

PahoMqttClient::~PahoMqttClient() {
    deinit(this);
}

int PahoMqttClient::connect()
{
    RETURN_ERR_IF_UNINITIALIZED
    MQTTClient_connectOptions opts = MQTTClient_connectOptions_initializer;
    opts.keepAliveInterval = this->conn_opts.keepAliveInterval;
    opts.cleansession = this->conn_opts.cleansession;
    opts.username = this->conn_opts.username;
    opts.password = this->conn_opts.password;
    COM_PRINTF_DEBUG("connect to server: %s\n", this->serverURI);
    int code= MQTTClient_connect(client, &opts);
    if (code != MQTTCLIENT_SUCCESS)
    {
        set_local_error_message(code);
        return THMR_CONNECT_FAILED;
    }
    return THMR_OK;
}

void PahoMqttClient::disconnect() {
    RETURN_IF_UNINITIALIZED
    MQTTClient_disconnect(client, 10000);
}

// 同步发送消息
// 等待发布完成返回 THMR_OK, 否则返回 THMR_PUBLISH_FAILED
int PahoMqttClient::publish(const char* topic, const char* message) {
    RETURN_ERR_IF_UNINITIALIZED
    wait_for_conn(this);
    MQTTClient_message pubmsg = MQTTClient_message_initializer;
    MQTTClient_deliveryToken token;
    pubmsg.payload = (void*)message;
    pubmsg.payloadlen = strlen(message);
    pubmsg.qos = this->qos;
    pubmsg.retained = 0;
    COM_PRINTF_DEBUG("publish topic: '%s', message: %s\n", topic, message);
    pthread_mutex_lock(&publish_mutex);
    int code = MQTTClient_publishMessage(client, topic, &pubmsg, &token);
    pthread_mutex_unlock(&publish_mutex);
    if(code != MQTTCLIENT_SUCCESS){
        set_local_error_message(code);
        return THMR_PUBLISH_FAILED;
    }
    // 等待发布完成
    code = MQTTClient_waitForCompletion(client,token, conn_opts.connectTimeout*1000);
    if(code != MQTTCLIENT_SUCCESS){
        set_local_error_message(code);
        return THMR_PUBLISH_FAILED;
    }
    return THMR_OK;
}

int PahoMqttClient::subscribe(const char* topic) {
    RETURN_ERR_IF_UNINITIALIZED
    COM_PRINTF_DEBUG("subscribe topic: %s\n", topic);
    wait_for_conn(this);
    pthread_mutex_lock(&subscribe_mutex);
    int code = MQTTClient_subscribe(client, topic, qos);
    pthread_mutex_unlock(&subscribe_mutex);
    if (code != MQTTCLIENT_SUCCESS) {
        set_local_error_message(code);
        return THMR_SUBSCRIBE_FAILED;
    }
    return THMR_OK;
}

int PahoMqttClient::unsubscribe(const char *channel)
{
    RETURN_ERR_IF_UNINITIALIZED
    pthread_mutex_lock(&subscribe_mutex);
    int code = MQTTClient_unsubscribe(client, channel);
    pthread_mutex_unlock(&subscribe_mutex);
    if (code != MQTTCLIENT_SUCCESS) {
        set_local_error_message(code);
        return THMR_UNSUBSCRIBE_FAILED;
    }
    return THMR_OK;
}

} // namespace smq

THMAR_API paho_mqtt_client_t thmar_mqtt_client_create0(const MQTTClient_connectOptions *conn_opts, const char *serverURI, const char *clientID, int qos)
{
    if (conn_opts == nullptr)
    {
        return nullptr;
    }
    smq::PahoMqttClient *p = new smq::PahoMqttClient(*conn_opts, serverURI, clientID, qos);
    return (paho_mqtt_client_t)p;
}

THMAR_API paho_mqtt_client_t thmar_mqtt_client_create(const char *serverURI, const char *username, const char *password, const char *clientID, int timeout, int keepAliveInterval, int qos)
{
    smq::PahoMqttClient *p = new smq::PahoMqttClient(serverURI, username, password, clientID, timeout, keepAliveInterval, qos);
    return (paho_mqtt_client_t)p;
}

THMAR_API void thmar_mqtt_client_destroy(paho_mqtt_client_t pahoClient)
{
    if (pahoClient)
    {
        smq::PahoMqttClient *p = (smq::PahoMqttClient *)pahoClient;
        delete p;
    }
}

THMAR_API int thmar_mqtt_client_connect(paho_mqtt_client_t pahoClient)
{
    if (pahoClient == nullptr)
    {
        return THMR_NULL_POINTER;
    }
    smq::PahoMqttClient *p = (smq::PahoMqttClient *)pahoClient;
    return p->connect();
}

THMAR_API void thmar_mqtt_client_disconnect(paho_mqtt_client_t pahoClient)
{
    if (pahoClient == nullptr)
    {
        return;
    }
    smq::PahoMqttClient *p = (smq::PahoMqttClient *)pahoClient;
    p->disconnect();
}

THMAR_API int thmar_mqtt_client_publish(paho_mqtt_client_t pahoClient, const char *topic, const char *message)
{
    if (pahoClient == nullptr || topic == nullptr || message == nullptr)
    {
        return THMR_NULL_POINTER;
    }
    smq::PahoMqttClient *p = (smq::PahoMqttClient *)pahoClient;
    return p->publish(topic, message);
}

THMAR_API int thmar_mqtt_client_subscribe(paho_mqtt_client_t pahoClient, const char *topic)
{
    if (pahoClient == nullptr || topic == nullptr)
    {
        return THMR_NULL_POINTER;
    }
    smq::PahoMqttClient *p = (smq::PahoMqttClient *)pahoClient;
    return p->subscribe(topic);
}

THMAR_API int thmar_mqtt_client_unsubscribe(paho_mqtt_client_t pahoClient, const char *channel)
{
    if (pahoClient == nullptr || channel == nullptr)
    {
        return THMR_NULL_POINTER;
    }
    smq::PahoMqttClient *p = (smq::PahoMqttClient *)pahoClient;
    return p->unsubscribe(channel);
}
